package com.shujia.wyh.hdfs;

/*
    1、获取配置文件对象（hadoop内置了一个类给我们创建对象）
    2、指定远程hadoop集群的链接
    3、创建集群操作文件对象 需要借助hadoop内置的一个类来获取FileSystem
 */

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;

import java.io.IOException;
import java.io.InputStream;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.Arrays;

public class HDFSDemo1 {
    public static void main(String[] args) {
        //1、获取配置文件对象（hadoop内置了一个类给我们创建对象）
        Configuration conf = new Configuration();

        //2、指定远程hadoop集群的链接
        conf.set("fs.defaultFS", "hdfs://master:9000");
        FileSystem fs = null;

        //通过上面的配置文件专门获取hdfs操作对象
        try {
            fs = FileSystem.get(conf);

            // 创建文件夹操作
//            Path path = new Path("/shujia/test1");
//
//            boolean b1 = fs.mkdirs(path);
//            if(b1){
//                System.out.println("文件夹创建成功！");
//            }else {
//                System.out.println("文件夹创建失败！");
//            }

            // 上传文件操作
            // 路径：本地文件 hadoop/data/students.txt
            // 路径：目标路径 /shujia/test1/
//            Path path1 = new Path("hadoop/data/students.txt");
//            Path path2 = new Path("/shujia/test1/");
//            fs.copyFromLocalFile(path1,path2);
//            System.out.println("文件上传成功！");


            //修改副本数
//            Path path = new Path("/shujia/test1/students.txt");
//            short s1 = 1;
//            boolean b = fs.setReplication(path, s1);
//            if(b){
//                System.out.println("副本数修改成功！！");
//            }else {
//                System.out.println("副本数修改失败！！");
//            }

            //获取文件基本信息
//            Path path = new Path("/shujia/test1/students.txt");
//            FileStatus fileStatus = fs.getFileStatus(path);
//            System.out.println(fileStatus.getPermission());
//            System.out.println(fileStatus.getGroup());
//            System.out.println(fileStatus.getBlockSize());

            //查看文件block块的信息
//            Path path = new Path("/shujia/test1/students.txt");
//            FileStatus fileStatus = fs.getFileStatus(path);
//            //获取整个文件所有的block块信息
//            BlockLocation[] blockLocations = fs.getFileBlockLocations(path, 0, fileStatus.getLen());
//            for (BlockLocation blockLocation : blockLocations) {
//                System.out.println("存储的节点：" + Arrays.toString(blockLocation.getNames()));
//                System.out.println("存储的主机名：" + Arrays.toString(blockLocation.getHosts()));
//                System.out.println("偏移量：" + blockLocation.getOffset());
//            }

            //下载文件 需要本地配置一下hadoop环境
//            Path path1 = new Path("/shujia/test1/students.txt");
//            Path path2 = new Path("hadoop/data2/students.txt");
//            fs.copyToLocalFile(path1,path2);


            //第二种方式上传文件到hadoop集群
            //对流复制的方式
            //获取本地的路径
            String localFilePath = "hadoop/data/students.txt";
            String hdfsFilePath = "/shujia/test2/students.txt";

            //打开本地文件输入流
            InputStream inputStream = Files.newInputStream(Paths.get(localFilePath));

            //创建HDFS文件输出流
            FSDataOutputStream fsDataOutputStream = fs.create(new Path(hdfsFilePath));

            //下载文件，打开hdfs上的文件
//            FSDataInputStream xxx = fs.open(new Path("xxx"));
//            IOUtils.copy(xxx,改成自己的本地路径，可以不存在);
            //TODO:自己使用对流复制完成下载文件

            //使用一个工具类来完成对流复制
            IOUtils.copy(inputStream,fsDataOutputStream);

            //关闭输入流和输出流
            IOUtils.closeQuietly(fsDataOutputStream);
            IOUtils.closeQuietly(inputStream);



        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            if(fs!=null){
                try {
                    fs.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

    }
}
